/* autor Konjkov.VV@gmail.com */

#include <string>
#include <iostream>
#include <sstream>
#include <cstring>

#include "common.h"
#include "Database.h"
#include "Syncctl.h"
#include "Packet.h"

Syncctl::Syncctl()
{
    usage = "Usage";
    conf = Syncconf();
};

void Syncctl::print_usage()
{
    std::cout << usage << std::endl;
}

void Syncctl::system_install()
{
// create schema SQL_SCHEMA;
// create language "plpgsql";

std::string SQL_SYSTEM_TABLE1=\
"CREATE TABLE "+ conf.schemaname +".syncMAIN ( \
version_major smallint not null, \
version_minor smallint not null, \
current_packet int default 0 not null)";

std::string SQL_SYSTEM_TABLE2=\
"CREATE TABLE "+ conf.schemaname +".syncNODES ( \
name varchar(30) not null unique, \
prepared int default 0 not null, \
updated int default 0 not null)";

std::string SQL_SYSTEM_TABLE3=\
"CREATE TABLE "+ conf.schemaname +".syncSTREAM ( \
sender varchar(30) not null, \
receiver varchar(30) not null, \
stage int not null, \
num_packet int not null, \
time_packet timestamp default current_timestamp not null, \
primary key (sender, receiver, stage, num_packet) )";

std::string SQL_SYSTEM_TABLE4=\
"CREATE TABLE "+ conf.schemaname +".syncPROCESSING ( \
sender varchar(30) not null, \
receiver varchar(30) not null, \
num_packet int not null, \
dt_receive timestamp, \
dt_process timestamp, \
st_process int default 0, \
primary key (sender, receiver, num_packet))";

std::string SQL_FUNCTION_SYNC_STREAM=\
"CREATE OR REPLACE FUNCTION "+ conf.schemaname +".syncNODES_Stream_Func_AU() \
RETURNS trigger AS \
\$BODY\$ \
BEGIN \
if "+SQL_FIELDPREFIX_OLD+".prepared < "+SQL_FIELDPREFIX_NEW+".prepared \
then \
    INSERT INTO "+ conf.schemaname +".syncSTREAM(sender, receiver, stage, num_packet, time_packet) \
    VALUES("+ conf.nodename +", "+SQL_FIELDPREFIX_NEW+".name, 0, "+SQL_FIELDPREFIX_NEW+".prepared, now()); \
end if; \
if "+SQL_FIELDPREFIX_OLD+".updated < "+SQL_FIELDPREFIX_NEW+".updated \
then \
    INSERT INTO "+ conf.schemaname +".syncSTREAM(sender, receiver, stage, num_packet, time_packet) \
    VALUES("+SQL_FIELDPREFIX_NEW+".name, "+ conf.nodename +", 0, "+SQL_FIELDPREFIX_NEW+".updated, now()); \
end if; \
RETURN NULL; \
END; \
\$BODY\$ \
LANGUAGE 'plpgsql'";

std::string SQL_TRIGGER_SYNC_STREAM=\
"CREATE TRIGGER syncNODES_Stream_AU \
AFTER UPDATE on "+ conf.schemaname +".syncNODES \
FOR EACH ROW EXECUTE PROCEDURE "+ conf.schemaname +".syncNODES_Stream_Func_AU()";

    Database d;
    d.Connect(conf.dbsource);
    d.AllocStatement().ExecDirect(SQL_SYSTEM_TABLE1);
    d.AllocStatement().ExecDirect(SQL_SYSTEM_TABLE2);
    d.AllocStatement().ExecDirect(SQL_SYSTEM_TABLE3);
    d.AllocStatement().ExecDirect(SQL_SYSTEM_TABLE4);
    d.AllocStatement().ExecDirect(SQL_FUNCTION_SYNC_STREAM);
    d.AllocStatement().ExecDirect(SQL_TRIGGER_SYNC_STREAM);
    d.AllocStatement().ExecDirect("INSERT INTO mcc.syncMAIN(version_major, version_minor) VALUES("+DATA_VER_MAJOR+","+DATA_VER_MINOR+")");
    d.Commit();
}

void Syncctl::system_uninstall()
{
    Database d;
    d.Connect(conf.dbsource);
    d.AllocStatement().ExecDirect("DROP TABLE mcc.syncMAIN");
    d.AllocStatement().ExecDirect("DROP TABLE mcc.syncNODES");
    d.AllocStatement().ExecDirect("DROP TABLE mcc.syncSTREAM");
    d.AllocStatement().ExecDirect("DROP TABLE mcc.syncPROCESSING");
    d.AllocStatement().ExecDirect("DROP TRIGGER mcc.syncNODES_Stream_AU()");
    d.AllocStatement().ExecDirect("DROP FUNCTION mcc.syncNODES_Stream_Func_AU()");
    d.Commit();
};

void Syncctl::system_update()
{
};

void Syncctl::system_status()
{
// Вернуть версию метаданных системы в БД и текущий номер пакета
std::string SQL_SYSTEM_STATUS_VERSION=\
"SELECT version_major, version_minor, current_packet FROM "+ conf.schemaname +".syncMAIN";

// Вернуть список удаленных узлов, зарегистрированных в системе
std::string SQL_SYSTEM_STATUS_NODES=\
"SELECT name, prepared, updated FROM "+ conf.schemaname +".syncNODES ORDER BY name";

// Вернуть список синхронизируемых таблиц
std::string SQL_TABLELIST=\
"SELECT DISTINCT table_schema || '.' || table_name \
FROM information_schema.columns \
WHERE column_name = '"+SQL_SYSFIELD1+"' \
AND table_name NOT LIKE '%_"+SQL_PREFIX_DELTABLE+"' \
AND table_name NOT LIKE '%_"+SQL_PREFIX_TEMPTABLE+"' \
AND table_name NOT LIKE '%_"+SQL_PREFIX_UNLOADTABLE+"'";

    Database d;
    d.Connect(conf.dbsource);

// Вернуть признак установлена ли система в базе данных.

    Statement s1 = d.AllocStatement();
    s1.Tables(conf.schemaname, "syncmain", "TABLE");
    if (!s1.Fetch()) {
        std::cout << "Система не установлена в базе данных" << std::endl;
    }

// Вернуть версию метаданных системы в БД и текущий номер пакета

    Statement s2 = d.AllocStatement();
    s2.ExecDirect(SQL_SYSTEM_STATUS_VERSION);
    if (s2.Fetch()) {
        std::cout << "Версия метаданных системы в БД: "
                  << (short)s2.col[0] << "."
                  << (short)s2.col[1] << std::endl;
        std::cout << "Текущий пакет: " << (int)s2.col[2] << std::endl;
    } else {
        std::cout << "Информация о версии системы отсутствует" << std::endl;
        std::cout << "Текущий номер пакета не указан" << std::endl;
    }

// Вернуть список удаленных узлов, зарегистрированных в системе

    std::cout << "Удаленные узлы:" << std::endl;

    Statement s3 = d.AllocStatement();
    s3.ExecDirect(SQL_SYSTEM_STATUS_NODES);
    while (s3.Fetch()) {
        std::cout << "\t" << s3.col[0] << ":["
                          << (int)s3.col[1] << ","
                          << (int)s3.col[2] << "]" << std::endl;
    }

// Вернуть список синхронизируемых таблиц

    std::cout << "Синхронизируемые таблицы:" << std::endl;

    Statement s4 = d.AllocStatement();
    s4.ExecDirect(SQL_TABLELIST);
    while (s4.Fetch()) {
        std::cout << "\t" << s4.col[0] << std::endl;
    }
    d.Commit();
};

void Syncctl::node_add(char* node)
{
// todo check in node is not alredy exist
// todo create subdirs
    Database d;
    d.Connect(conf.dbsource);
    d.AllocStatement().ExecDirect("INSERT INTO mcc.syncNODES(name) VALUES('"+std::string(node)+"')");
    d.Commit();
};

void Syncctl::node_remove(char* node)
{
// todo check in node is alredy exist
    Database d;
    d.Connect(conf.dbsource);
    d.AllocStatement().ExecDirect("DELETE FROM mcc.syncnodes WHERE name = '"+std::string(node)+"'");
    d.Commit();
};

void Syncctl::node_reset(char *node)
{
// todo check in node is alredy exist
//    prepared_set(node, 0);
// todo remove outgoing packets
};

void Syncctl::node_list()
{
//SQLHSTMT stmt;
// todo
};

void Syncctl::table_add(std::string name)
{
std::string owner;
std::string table;
std::stringstream allfields;
std::stringstream allvalues;
std::stringstream allfieldsdef;

    owner = name.substr(0,name.find("."));
    table = name.substr(name.find(".")+1,name.length());

// Триггер для определения номера пакета (SQL_SYSFIELD2),
// в который запись попала впервые.
std::string SQL_FUNCTION_BU1=\
"CREATE OR REPLACE FUNCTION "+owner+"."+table+"_sync_func_bu1() \
RETURNS trigger AS \
\$BODY\$ \
BEGIN \
if ("+SQL_FIELDPREFIX_NEW+"."+SQL_SYSFIELD1+" <> "+SQL_FIELDPREFIX_OLD+"."+SQL_SYSFIELD1+" \
    and "+SQL_FIELDPREFIX_NEW+"."+SQL_SYSFIELD1+" <> -1 \
    and "+SQL_FIELDPREFIX_OLD+"."+SQL_SYSFIELD2+" = 0) \
then \
    "+SQL_FIELDPREFIX_NEW+"."+SQL_SYSFIELD2+" := "+SQL_FIELDPREFIX_NEW+"."+SQL_SYSFIELD1+"; \
end if; \
RETURN NEW; \
END; \
\$BODY\$ \
LANGUAGE 'plpgsql'";

std::string SQL_TRIGGER_BU1=\
"CREATE TRIGGER "+table+"_sync_bu1 \
BEFORE UPDATE on "+owner+"."+table+" \
FOR EACH ROW \
EXECUTE PROCEDURE "+owner+"."+table+"_sync_func_bu1()";

// Триггер для отслеживания изменения записи.
// Триггер должен срабатывать после всех триггеров,
// вызываемых до изменения записи и только в случае,
// если изменились значения любых полей, кроме
// служебных (другими словами, изменение значения
// любого из служебных полей не должно приводить
// к вызыву данного триггера).

std::string SQL_FUNCTION_BU2=\
"CREATE OR REPLACE FUNCTION "+owner+"."+table+"_sync_func_bu2() \
RETURNS trigger AS \
\$BODY\$ \
BEGIN \
if ("+SQL_FIELDPREFIX_NEW+"."+SQL_SYSFIELD1+" = "+SQL_FIELDPREFIX_OLD+"."+SQL_SYSFIELD1+" \
    and "+SQL_FIELDPREFIX_NEW+"."+SQL_SYSFIELD2+" = "+SQL_FIELDPREFIX_OLD+"."+SQL_SYSFIELD2+") \
then \
    "+SQL_FIELDPREFIX_NEW+"."+SQL_SYSFIELD1+" := 0; \
end if; \
RETURN NEW; \
END; \
\$BODY\$ \
LANGUAGE 'plpgsql'";

    Database d;
    d.Connect(conf.dbsource);
    Statement s1 = d.AllocStatement();
    s1.Columns(owner, table, "");
    while (s1.Fetch()) {
        if (!(s1.col[3].asString() == "packet" || s1.col[3].asString() == "initpacket")) {
            allfields << s1.col[3] << ",";
            allvalues << SQL_FIELDPREFIX_OLD << "." << s1.col[3] << ",";
            switch((short)s1.col[4]) {
            case SQL_NUMERIC:
                allfieldsdef << s1.col[3] << " " << s1.col[5] << "(" << (int)s1.col[6] << "," << (short)s1.col[8] << "),";
                break;
            case SQL_CHAR:
                allfieldsdef << s1.col[3] << "char(" << (int)s1.col[6] << "),";
                break;
            case SQL_VARCHAR:
                allfieldsdef << s1.col[3] << "varchar(" << (int)s1.col[6] << "),";
                break;
            default:
                allfieldsdef << s1.col[3] << " " << s1.col[5] << ",";
                break;
            }
        }
    }
    std::cout << allfieldsdef.str() << std::endl;

std::string SQL_TRIGGER_BU2=\
"CREATE TRIGGER "+table+"_sync_bu2 \
BEFORE UPDATE on "+table+"."+owner+" \
FOR EACH ROW \
EXECUTE PROCEDURE "+table+"."+owner+"_sync_func_bu2()";

// SQL-скрипт создания таблицы для удаленных записей
// Параметры шаблона:
//   owner          - владелец таблицы
//   table          - имя таблицы
//   all_fields_def - определения всех полей таблицы (кроме служебных полей)
//                    из удаленной БД, перечисленные через запятую

std::string SQL_WATCHDELETE_TABLE=\
"CREATE TABLE "+owner+"."+table+"_"+SQL_PREFIX_DELTABLE+" ( \
"+allfieldsdef.str()+", \
"+SQL_SYSFIELD1+" integer DEFAULT 0 NOT NULL, \
"+SQL_SYSFIELD2+" integer DEFAULT 0 NOT NULL)";

// SQL-скрипт создания триггера на удаление записи
// Параметры шаблона:
//   owner          - владелец таблицы
//   table          - имя таблицы
//   all_fields     - список всех полей таблицы (кроме служебных полей),
//                    перечисленных через запятую
//   all_values     - список всех полей таблицы (кроме служебных полей)
//                    в виде:
//                      old_values.<field1>,
//                      old_values.<field2>,
//                      ...
//                      old_values.<fieldN>

std::string SQL_WATCHDELETE_FUNCTION=\
"CREATE OR REPLACE FUNCTION "+owner+"."+table+"_sync_func_ad() \
RETURNS trigger AS \
\$BODY\$ \
BEGIN \
if ("+SQL_FIELDPREFIX_OLD+"."+SQL_SYSFIELD2+" <> 0) then \
    INSERT INTO "+owner+"."+table+"_"+SQL_PREFIX_DELTABLE+"( \
        "+allfields.str()+", \
        "+SQL_SYSFIELD1+", \
        "+SQL_SYSFIELD2+") \
    VALUES( \
        "+allvalues.str()+", \
        0, \
        "+SQL_FIELDPREFIX_OLD+"."+SQL_SYSFIELD2+"); \
end if; \
RETURN NULL; \
END; \
\$BODY\$ \
LANGUAGE 'plpgsql'";

std::string SQL_WATCHDELETE_TRIGGER=\
"CREATE TRIGGER "+table+"_sync_ad \
AFTER DELETE on "+owner+"."+table+" \
FOR EACH ROW \
EXECUTE PROCEDURE "+owner+"."+table+"_sync_func_ad()";

//    table_remove(name);

    d.AllocStatement().ExecDirect("ALTER TABLE "+owner+"."+table+" ADD "+SQL_SYSFIELD1+" integer DEFAULT 0 NOT NULL");
    d.AllocStatement().ExecDirect("ALTER TABLE "+owner+"."+table+" ADD "+SQL_SYSFIELD2+" integer DEFAULT 0 NOT NULL");
    d.AllocStatement().ExecDirect("CREATE INDEX i_"+owner+"_"+table+"_"+SQL_SYSFIELD1+" ON "+owner+"."+table+"("+SQL_SYSFIELD1+")");
    d.AllocStatement().ExecDirect(SQL_FUNCTION_BU1);
    d.AllocStatement().ExecDirect(SQL_TRIGGER_BU1);
    d.AllocStatement().ExecDirect(SQL_FUNCTION_BU2);
    d.AllocStatement().ExecDirect(SQL_TRIGGER_BU2);
    d.AllocStatement().ExecDirect(SQL_WATCHDELETE_TABLE);
    d.AllocStatement().ExecDirect(SQL_WATCHDELETE_FUNCTION);
    d.AllocStatement().ExecDirect(SQL_WATCHDELETE_TRIGGER);
    d.Commit();
};

void Syncctl::table_remove(std::string name) {
std::string owner;
std::string table;

    owner = name.substr(0,name.find("."));
    table = name.substr(name.find(".")+1,name.length());

    Database d;
    d.Connect(conf.dbsource);

// SQL-скрипт выключения таблицы из процесса синхронизации
    d.AllocStatement().ExecDirect("DROP TRIGGER zzz_"+table+"_sync_ad ON "+owner+"."+table);
    d.AllocStatement().ExecDirect("DROP FUNCTION "+owner+"."+table+"_sync_func_ad()");
    d.AllocStatement().ExecDirect("DROP TRIGGER zz_"+table+"_sync_bu1 ON "+owner+"."+table);
    d.AllocStatement().ExecDirect("DROP FUNCTION "+owner+"."+table+"_sync_func_bu1()");
    d.AllocStatement().ExecDirect("DROP TRIGGER zzz_"+table+"_sync_bu2 ON "+owner+"."+table);
    d.AllocStatement().ExecDirect("DROP FUNCTION "+owner+"."+table+"_sync_func_bu2()");
    d.AllocStatement().ExecDirect("DROP INDEX "+owner+".i_"+owner+"_"+table+"_"+SQL_SYSFIELD1);
    d.AllocStatement().ExecDirect("ALTER TABLE "+owner+"."+table+" DROP "+SQL_SYSFIELD1);
    d.AllocStatement().ExecDirect("ALTER TABLE "+owner+"."+table+" DROP "+SQL_SYSFIELD2);
    d.AllocStatement().ExecDirect("DROP TABLE "+owner+"."+table+"_"+SQL_PREFIX_DELTABLE);
    d.AllocStatement().ExecDirect("DROP TABLE "+owner+"."+table+"_"+SQL_PREFIX_UNLOADTABLE);
    d.Commit();
};

void Syncctl::table_update(std::string name) {
std::string owner;
std::string table;
};

void Syncctl::table_dwatch_on(std::string name) {
std::string owner;
std::string table;

    owner = name.substr(0,name.find("."));
    table = name.substr(name.find(".")+1,name.length());

    Database d;
    d.Connect(conf.dbsource);

//    d.ExecDirect(SQL_WATCHDELETE_TABLE);
//    d.ExecDirect(SQL_WATCHDELETE_FUNCTION);
//    d.ExecDirect(SQL_WATCHDELETE_TRIGGER);
    d.Commit();
};

void Syncctl::table_dwatch_off(std::string name)
{
std::string owner;
std::string table;

    owner = name.substr(0,name.find("."));
    table = name.substr(name.find(".")+1,name.length());
};

void Syncctl::data_prepare(char *node) {

//struct nodesconf	conf;
//struct group*		nextgroup;
//struct nodesection*	nextnode;
//struct nodename*	nextnodename;
//struct nodetable*	nexttable;

    OutPacket p(node);

//    nodesconf(&conf);
//    nextnode=conf.nodessection;
//    while(nextnode!=NULL) {
//        nextnodename=nextnode->nodes;
//        nexttable=nextnode->tables;
//        while(nextnodename!=NULL) {
//            printf("%s\n",nextnodename->name);
//            if (strcmp(nextnodename->name, node)==0) {
//                while(nexttable!=NULL) {

                    p.addTable("public.mcs_sale","*","","public.mcs_sales");

//                    nexttable=nexttable->next;
//                }
//            }
//            nextnodename=nextnodename->next;
//        }
//        nextnode=nextnode->next;
//    }

//    prepared_set(node, curpacket+1);

// SQL-скрипт для формирования пакета - КОНЕЦ
// Если были выгруженны хоть какие-либо данные
std::string SQL_DATA_PREPARE_AFTER = "UPDATE "+ conf.schemaname + ".syncMAIN SET current_packet = ?";
//    sprintf( buffer, SQL_DATA_PREPARE_AFTER, schema);
//    SQLAllocStmt( dbc, &stmt );
//    EXEC_SQL( SQLBindParameter( stmt, 1, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_CHAR, 50, 0, curpacket+1, 0, SQL_TYPE_NULL) );
//    EXEC_SQL( SQLExecDirect( stmt, buffer, SQL_NTS ) );
//    SQLFreeStmt( stmt, SQL_CLOSE );

//    packetend(packet_info);

// SQL-скрипт для формирования пакета - ДЛЯ ОТДЕЛЬНОГО ПРЕДСТАВЛЕНИЯ
// Параметры шаблона как для таблицы, плюс:
//   select_special  - названия всех служебных полей в виде списка:
//                       pkt_<field1> as ${SQL_FIELDPREFIX1}pkt_<field1>
//                       pkt_<field2> as ${SQL_FIELDPREFIX1}pkt_<field2>
//                       ...
//                       pkt_<fieldN> as ${SQL_FIELDPREFIX1}pkt_<fieldN>
//   changed_special - условие на изменение хотя бы одного из служебного
//                     поля (на основе шаблона SQL_DATA_PREPARE_VIEW_IF)
//   update_special  - обновление всех служебных полей (на основе шаблона
//                     SQL_DATA_PREPARE_VIEW_UPDATE)

std::string SQL_DATA_PREPARE_VIEW = "SELECT <select_fields> FROM <owner>.<table> WHERE ( <changed_special> ) <filter>";
//  <update_special>

// SQL-скрипт для формирования пакета - ДЛЯ ОТДЕЛЬНОГО ПРЕДСТАВЛЕНИЯ
// Условие на изменение отдельного служебного поля

std::string SQL_DATA_PREPARE_VIEW_IF = "<special_field> = 0 or <special_field> > (SELECT prepared FROM <schema>.syncNODES WHERE name = '<node>')";

// SQL-скрипт для формирования пакета - ДЛЯ ОТДЕЛЬНОГО ПРЕДСТАВЛЕНИЯ
// Для изменения отдельного служебного поля

std::string SQL_DATA_PREPARE_VIEW_UPDATE = "UPDATE <owner>.<table> SET <special_field> = %s WHERE <special_field> = 0";

}

void Syncctl::data_update(char* node)
{
    InPacket p(node);
};

void Syncctl::prepared_set(char* node, char* number)
{
// Установить номер последнего успешно сформированного пакета
// для определенного удаленного узла
    Database d;
    d.Connect(conf.dbsource);
    d.AllocStatement().ExecDirect("UPDATE "+conf.schemaname+".syncNODES SET prepared = '"+number+"' WHERE name = '"+node+"'");
    d.Commit();
}

int Syncctl::prepared_get(char* node)
{
// Вернуть номер последнего успешно сформированного пакета
// для определенного удаленного узла
    Database d;
    d.Connect(conf.dbsource);
    Statement s = d.AllocStatement();
    s.ExecDirect("SELECT prepared FROM "+conf.schemaname+".syncNODES WHERE name = '"+node+"'");
    if (s.Fetch()) {
        return s.col[0];
    }
}

void Syncctl::updated_set(char* node, char* number)
{
// Установить номер последнего успешно обработанного пакета
// для определенного удаленного узла
    Database d;
    d.Connect(conf.dbsource);
    d.AllocStatement().ExecDirect("UPDATE "+conf.schemaname+".syncNODES SET updated = '"+number+"' WHERE name = '"+node+"'");
    d.Commit();
}

int Syncctl::updated_get(char* node)
{
// Вернуть номер последнего успешно обработанного пакета
// для определенного удаленного узла
    Database d;
    d.Connect(conf.dbsource);
    Statement s = d.AllocStatement();
    s.ExecDirect("SELECT updated FROM "+conf.schemaname+".syncNODES WHERE name = '"+node+"'");
    if (s.Fetch()) {
        return s.col[0];
    }
}

Syncctl::~Syncctl(){
};
